package com.example.mq1.mqserver;

import com.example.mq1.common.Consumer;
import com.example.mq1.common.MqException;
import com.example.mq1.common.Router;
import com.example.mq1.mqserver.core.*;
import com.example.mq1.mqserver.datacenter.DiskDataCenter;
import com.example.mq1.mqserver.datacenter.MemoryDataCenter;
import com.sun.java.accessibility.util.EventID;

import javax.websocket.RemoteEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class VirtualHost {

    private String virtualHostName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    //解决线程安全问题
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();


    /**
     * 构造
     */
    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;

        //对硬盘数据进行初始化，加载到数据库中
        diskDataCenter.init();
        //内存不需要初始化，只需要恢复数据即可(将硬盘上的数据加载到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 初始化加载数据失败");
        }
    }

    //只需要 getter 即可
    public String getVirtualHostName() {
        return virtualHostName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    /**
     * 创建交换机
     * @param exchangeName
     * @param exchangeType
     * @param durable
     * @param autoDelete
     * @param arguments
     * @return
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,
                                    boolean autoDelete, Map<String, Object> arguments) {
        //1.虚拟主机隔离
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                //2.检查内存是否存在该虚拟主机
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if(exchange != null) {
                    System.out.println("[VirtualHost] 该交换机已存在！ exchangeName=" + exchangeName);
                    return true;
                }
                //3.不存在就创建
                exchange = new Exchange();
                exchange.setExchangeName(exchangeName);
                exchange.setExchangeType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //4.内存/硬盘存储 (这里要先写硬盘，因为硬盘最容易引发异常，如果先写内存，写入成功后还需要写硬盘，硬盘一旦发生异常，内存中的就需要清理，不方便操作)
                if(durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除交换机
     * @param exchangeName
     */
    public boolean exchangeDelete(String exchangeName) {
        //1.虚拟主机隔离
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                //2.判断当前交换机是否存在
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if(exchange == null) {
                    System.out.println("[VirtualHost] 当前交换机不存在！ exchangeName=" + exchangeName);
                }
                //3.存在就删除(内存/硬盘)
                if(exchange.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！ exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败！ exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建队列
     * @param queueName
     * @param durable
     * @param exclusive
     * @param autoDelete
     * @param arguments
     * @return
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive,
                                boolean autoDelete, Map<String, Object> arguments) {
        //1.虚拟主机隔离
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //2.检查当前是否存在队列
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue != null) {
                    System.out.println("[VirtualHost] 当前队列已存在！ queueName=" + queueName);
                    return true;
                }
                //3.不存在就创建
                queue = new MSGQueue();
                queue.setQueueName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //4.内存/硬盘存储
                if(durable) {
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功！ queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败！ queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDelete(String queueName) {
        //1.虚拟主机隔离
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //2.判断当前队列是否存在
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null) {
                    throw new MqException("[VirtualHost] 当前队列不存在！ queueName=" + queueName);
                }
                //3.存在就删除(内存/硬盘)
                if(queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功！ queueName=" + queueName);
            }
            return true;
        } catch(Exception e) {
            System.out.println("[VirtualHost] 队列删除失败！ queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建绑定
     * @param exchangeName
     * @param queueName
     * @param bindingKey
     * @return
     */
    public boolean queueBind(String exchangeName, String queueName, String bindingKey) {
        //1.虚拟主机隔离
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
               synchronized (queueLocker) {
                   //2.判断是否存在当前绑定
                   Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                   if(binding != null) {
                       System.out.println("[VirtualHost] 当前绑定存在！ exchangeName=" + exchangeName +
                               ", queueName=" + queueName);
                       return true;
                   }
                   //3.判断 bindingKey 是否合法
                   if(!router.checkBindingKey(bindingKey)) {
                       throw new MqException("[VirtualHost] 创建绑定失败，当前 bindingKey 不合法！bindingKey=" + bindingKey);
                   }
                   //4.判断交换机和队列是否存在
                   Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                   if(exchange == null) {
                       throw new MqException("[VirtualHost] 创建绑定失败，当前不存在该交换机！ exchangeName=" + exchangeName +
                               ", queueName=" + queueName);
                   }
                   //5.不存在绑定就创建
                   binding = new Binding();
                   binding.setExchangeName(exchangeName);
                   binding.setQueueName(queueName);
                   binding.setBindingKey(bindingKey);
                   MSGQueue queue = memoryDataCenter.getQueue(queueName);
                   if(queue == null) {
                       throw new MqException("[VirtualHost] 创建绑定失败，当前不存在该队列！ exchangeName=" + exchangeName +
                               ", queueName=" + queueName);
                   }
                   //6.写硬盘(exchange 和 queue 都持久化才写)
                   if(exchange.isDurable() && queue.isDurable()) {
                       diskDataCenter.insertBinding(binding);
                   }
                   //5.写内存
                   memoryDataCenter.insertBinding(binding);
                   System.out.println("[VirtualHost] 创建绑定成功！ exchangeName=" + exchangeName +
                           ", queueName=" + queueName);
               }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 创建绑定失败！ exchangeName=" + exchangeName +
                    ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 删除绑定
     * @param exchangeName
     * @param queueName
     * @return
     */
    public boolean queueUnBind(String exchangeName, String queueName) {
        //1.虚拟主机隔离
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //2.判断当前绑定是否存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(binding == null) {
                        throw new MqException("[VirtualHost] 绑定删除失败，不存在当前绑定！ exchangeName=" + exchangeName +
                                ", queueName=" + queueName);
                    }
                    //3.删除硬盘(这里无需在判断交换机和队列是否持久化，因为这里的删除本质是 sql 的 delete，为空删除也不会有什么副作用)
                    diskDataCenter.deleteBinding(binding);
                    //4.删除内存
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 绑定删除成功！ exchangeName=" + exchangeName +
                            ", queueName=" + queueName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定删除失败！ exchangeName=" + exchangeName +
                    ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 通过交换机将消息发送给指定队列(保存到内存/硬盘上)
     * @param exchangeName
     * @param routingKey
     * @param basicProperties
     * @param body
     * @return
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        //1.虚拟主机隔离
        exchangeName = virtualHostName + exchangeName;
        try {
            //2.判断 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] 消息发布失败，routingKey 不合法！ routingKey=" + routingKey);
            }
            //3.获取交换机
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null) {
                throw new MqException("[VirtualHost] 消息发布失败，交换机不存在！ exchangeName=" + exchangeName);
            }
            //4.根据不同交换机类型，进行消息的发送
            if(exchange.getExchangeType() == ExchangeType.DRICT) {
                //1) 直接交换机: 以队列名为 routingKey，直接匹配对应队列
                String queueName = virtualHostName + routingKey;
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null) {
                    throw new MqException("[VirtualHost] 消息发布失败，对应队列为空！ queueName=" + queueName);
                }
                sendMessage(queue, message);
            } else {
                //2) 删除交换机和主题交换机: 首先都要遍历所有的 binding，扇出交换机是每一个队列都要发一份，主题交换机是匹配对应的队列才发
                ConcurrentHashMap<String, Binding> bindingMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String, Binding> entry : bindingMap.entrySet()) {
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null) {
                        //这里不抛异常类，不然影响查找下一个 队列
                        System.out.println("[VirtualHost] 当前队列不存在！queueName=" + queue.getQueueName());
                    }
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    if(!router.route(exchange.getExchangeType(), binding, message)) {
                        continue;
                    }
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发布失败！ exchangeName=" + exchangeName +
                    ", routingKey=" + routingKey);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 将消息存储到对应队列的内存/硬盘中(deliverMode == 0x1 表示不需要 , 0x2 表示需要)
     * 最后通知消费者消费消息
     * @param queue
     * @param message
     */
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode = message.getDeliverMode();
        if(deliverMode == 2) {
            //需要持久化
            diskDataCenter.sendMessage(queue, message);
        }
        memoryDataCenter.sendMessage(queue, message);
        //通知消费者可以消费消息了
        consumerManager.notifyConsumer(queue.getQueueName());
    }

    /**
     * 消费者订阅队列
     * @param consumerTag
     * @param queueName
     * @param autoAck
     * @param consumer
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //1.虚拟主机隔离
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] 消费者订阅队列成功！consumerTag=" + consumerTag +
                    ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消费者订阅队列失败！consumerTag=" + consumerTag +
                    ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 主动应答
     * @param queueName
     * @param messageId
     * @return
     */
    public boolean basicAck(String queueName, String messageId) {
        //1.虚拟主机隔离
        queueName = virtualHostName + queueName;
        try {
            //2.获取队列
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null) {
                throw new MqException("[VirtualHost] 当前队列不存在！ queueName=" + queueName);
            }
            //3.获取消息
            Message message = memoryDataCenter.getMessages(messageId);
            //4.删除 硬盘/内存 消息
            if(message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            memoryDataCenter.removeMessage(message.getMessageId());
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] 主动应答成功！queueName=" + queueName +
                    ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 主动应答失败！queueName=" + queueName +
                    ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }

}
